//package com.java.springboot.elasticsearch.canal;
//
//import com.alibaba.fastjson.JSON;
//import com.alibaba.otter.canal.client.CanalConnector;
//import com.alibaba.otter.canal.protocol.CanalEntry;
//import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
//import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
//import com.alibaba.otter.canal.protocol.Message;
//import com.google.protobuf.InvalidProtocolBufferException;
//import com.java.springboot.elasticsearch.mapper.ReadBookPdMapper;
//import com.java.springboot.elasticsearch.model.ReadBookPd;
//import lombok.extern.slf4j.Slf4j;
//import org.elasticsearch.action.delete.DeleteRequest;
//import org.elasticsearch.action.index.IndexRequest;
//import org.elasticsearch.client.RequestOptions;
//import org.elasticsearch.client.RestHighLevelClient;
//import org.elasticsearch.common.xcontent.XContentType;
//import org.springframework.beans.BeansException;
//import org.springframework.context.ApplicationContext;
//import org.springframework.context.ApplicationContextAware;
//import org.springframework.scheduling.annotation.Scheduled;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.Resource;
//import java.io.IOException;
//import java.util.HashMap;
//import java.util.List;
//import java.util.Map;
//
///**
// * @author zhangtengfei
// * @date 2021/10/13 13:44
// */
//@Component
//@Slf4j
//public class CanalScheduling implements Runnable, ApplicationContextAware {
//
//    private ApplicationContext applicationContext;
//
//    @Resource
//    private CanalConnector canalConnector;
//
//    @Resource
//    private ReadBookPdMapper readBookPdMapper;
//
//    @Resource
//    private RestHighLevelClient restHighLevelClient;
//
//    @Override
//    @Scheduled(fixedDelay = 1000)
//    public void run() {
//        long batchId = -1;
//        try {
//            int batchSize = 1000;
//            // 获取指定数量的数据
//            Message message = canalConnector.getWithoutAck(batchSize);
//            // 获取批量ID
//            batchId = message.getId();
//            List<CanalEntry.Entry> entries = message.getEntries();
//            if (batchId != -1 && entries.size() > 0) {
//                entries.forEach(entry -> {
//                    // mysql中一行数据有变化时，EntryType有三种类型：TRANSACTIONBEGIN ROWDATA TRANSACTIONEND
//                    if (entry.getEntryType() == EntryType.ROWDATA) {
//                        // 解析处理
//                        publishCanalEvent(entry);
//                    }
//                });
//            }
//            //提交确认消费完毕
//            canalConnector.ack(batchId);
//        } catch (Exception e) {
//            e.printStackTrace();
//            //失败的话进行回滚
//            canalConnector.rollback(batchId);
//        }
//    }
//
//    private void publishCanalEvent(CanalEntry.Entry entry) {
//        log.info("收到canal消息: {}", entry.toString());
//        // 不关心 TRANSACTIONBEGIN TRANSACTIONEND 的类型
//        if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
//            return;
//        }
//        // 监听到的数据库名称
//        String database = entry.getHeader().getSchemaName();
//        // 变更的数据表名称
//        String table = entry.getHeader().getTableName();
//        CanalEntry.RowChange change;
//        try {
//            change = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
////            log.info("change: {}",change.toString());
//        } catch (InvalidProtocolBufferException e) {
//            e.printStackTrace();
//            return;
//        }
//        EventType eventType = change.getEventType();
//        change.getRowDatasList().forEach(rowData -> {
//            List<CanalEntry.Column> columns;
//            //对于es来说 只要关注 delete 和 update 还有insert
//            if (eventType == EventType.DELETE) {
//                columns = rowData.getBeforeColumnsList();
//            } else {
//                columns = rowData.getAfterColumnsList();
//            }
//            //解析成map 格式
//            Map<String, Object> dataMap = parseColumnsToMap(columns);
//            try {
//                //真正的去改es
//                indexES(dataMap, database, table, eventType);
//            } catch (IOException e) {
//                e.printStackTrace();
//            }
//        });
//    }
//
//    Map<String, Object> parseColumnsToMap(List<CanalEntry.Column> columns) {
//        Map<String, Object> jsonMap = new HashMap<>();
//        columns.forEach(column -> {
//            if (column == null) {
//                return;
//            }
//            jsonMap.put(column.getName(), column.getValue());
//        });
//        return jsonMap;
//    }
//
//    private void indexES(Map<String, Object> dataMap, String database, String table, EventType eventType)
//            throws IOException {
//        try {
//            if (eventType == EventType.DELETE) {
//                log.info("删除索引Id: {},type: {},value: {}", dataMap.get("id"), eventType.toString(), dataMap.toString());
//                DeleteRequest deleteRequest = new DeleteRequest("book", String.valueOf(dataMap.get("id")));
//                restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
//            } else {
//                //这里又两种i方式,一种是直接拿canal过来的数据，还有一种就是拿主键id去查询。
////                List<Map<String, Object>> result = readBookPdMapper.buildESQuery(new Integer((String) dataMap.get("id")));
//                ReadBookPd readBookPd = readBookPdMapper.selectByPrimaryKey(Integer.valueOf((String) dataMap.get("id")));
//                log.info("更新索引Id: {},type: {},value: {}", dataMap.get("id"), eventType.toString(), readBookPd.toString());
//
//                IndexRequest indexRequest = new IndexRequest("book");
//                indexRequest.id(String.valueOf(dataMap.get("id")));
//                indexRequest.source(JSON.toJSONString(readBookPd), XContentType.JSON);
//                restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
//
//                //处理多业务的思路
//                //1.监听的是商品表 变更时我会拿到商品id
//                //2.根据goodsId 去营销中心 订单中心 查询数据 会调用他们的接口
//                //3.组装数据 进入es
//            }
//
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//
//    }
//
//    @Override
//    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
//        this.applicationContext = applicationContext;
//    }
//}
//
